package avicit.bdp.dds.server.worker.adapter.yarn;

import avicit.bdp.core.constant.Constants;
import avicit.bdp.dds.server.entity.YarnSubmitConditions;
import com.alibaba.fastjson2.JSON;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author DIGITAL-MAYANJ
 */
public abstract class YarnAdapter {

    private static final Logger logger = LoggerFactory.getLogger(YarnAdapter.class);

    /**
     * 提交任务
     *
     * @param conditions 参数
     * @return 任务ID
     */
    public String submitTask(YarnSubmitConditions conditions) {
        return null;
    }

    /**
     * 停止任务
     *
     * @param yarnResourcemanagerAddress yarn资源管理器地址， 例如：master:8032，查看yarn集群获取具体地址
     * @param appIdStr                   需要取消的任务id
     */
    public void killJob(String yarnResourcemanagerAddress, String appIdStr) {

    }

    /**
     * 获取任务状态
     *
     * @param yarnResourcemanagerAddress yarn资源管理器地址， 例如：master:8032，查看yarn集群获取具体地址
     * @param appIdStr                   需要取消的任务id
     */
    public ApplicationReport getStatus(String yarnResourcemanagerAddress, String appIdStr) {
        logger.info("获取任务状态启动，任务id：{}", appIdStr);
        // 初始化 yarn的配置
        Configuration cf = new Configuration();
        String os = System.getProperty("os.name");
        boolean crossPlatform = false;
        if (os.contains(Constants.OS_WINDOWS)) {
            crossPlatform = true;
        }
        // 配置使用跨平台提交任务
        cf.setBoolean("mapreduce.app-submission.cross-platform", crossPlatform);
        // 设置yarn资源，不然会使用localhost:8032
        // 启用yarn的高可用，默认关闭
        cf.setBoolean("yarn.resourcemanager.ha.enabled", true);
        cf.set("yarn.resourcemanager.ha.rm-ids", "rm1,rm2");
        // 设置yarn资源，不然会使用localhost:8032
        // todo
        //String[] hostPort = RegexUtil.getResourceHostPort(yarnResourcemanagerAddress);
        String[] hostPort = null;
        cf.set("yarn.resourcemanager.address.rm1", hostPort[0]);
        // 设置yarn资源，不然会使用localhost:8032
        if (hostPort.length >= 2) {
            cf.set("yarn.resourcemanager.address.rm2", hostPort[1]);
        }
        logger.info("获取任务状态，任务id: {}", appIdStr);

        YarnClient yarnClient = YarnClient.createYarnClient();
        // 初始化yarn的客户端
        yarnClient.init(cf);
        // yarn客户端启动
        yarnClient.start();
        ApplicationReport report = null;
        try {
            report = yarnClient.getApplicationReport(getAppId(appIdStr));
        } catch (Exception e) {
            logger.error("获取任务状态失败", e);
        }

        // 关闭yarn客户端
        yarnClient.stop();
        logger.info("获取任务状态结束，任务状态： {}", JSON.toJSONString(report));
        return report;
    }

    /**
     * 根据任务id字符串获取对象
     *
     * @param appIdStr String类型的id
     * @return 应用id的对象
     */
    public ApplicationId getAppId(String appIdStr) {
        return ConverterUtils.toApplicationId(appIdStr);
    }

}
